package top.joylee.example.grpc;

import io.grpc.Channel;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.joylee.example.grpc.route.*;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class RouteGuideClient {
    private static final Logger log = LoggerFactory.getLogger(RouteGuideClient.class);
    private final Random random = new Random();
    // 阻塞stub
    private final RouteGuideGrpc.RouteGuideBlockingStub blockingStub;
    // 非阻塞stub
    private final RouteGuideGrpc.RouteGuideStub asyncStub;

    public RouteGuideClient(Channel channel) {
        blockingStub = RouteGuideGrpc.newBlockingStub(channel);
        asyncStub = RouteGuideGrpc.newStub(channel);
    }

    /**
     * 定义获取点位特征值的方法
     * 指定经度、纬度，拼装point message
     * 远程调用服务端，得到响应结果
     * 如果feature不为空而且feature中name的值不为空，则证明找到了
     * 否则，没找到该点位
     *
     * @param lat 经度
     * @param lon 纬度
     */
    public void getFeature(int lat, int lon) {
        log.info("***  Get Feature:[{},{}]", lat, lon);
        final Point request = Point.newBuilder()
                .setLatitude(lat)
                .setLongitude(lon)
                .build();

        //阻塞等待服务端响应
        final Feature feature = blockingStub.getFeature(request);
        if (Objects.nonNull(feature) && RouteGuideUtil.exists(feature)) {
            log.info("Found feature ,value = {}", feature);
        } else {
            log.info("Not found point,value = {}", request);
        }
    }

    /**
     * 通过指定“矩形”边界，来获取该边界内的特征点位
     * 瓶装 Rectangle message
     * 远程调用服务端，得到响应结果，并打印
     *
     * @param lowLat 低位经度
     * @param lowLon 低位纬度
     * @param hiLat  高位经度
     * @param hiLon  高位纬度
     */
    public void listFeatures(int lowLat, int lowLon, int hiLat, int hiLon) {
        log.info("*** ListFeatures: lowLat={} lowLon={} hiLat={} hiLon={}",
                lowLat, lowLon, hiLat, hiLon);

        final Rectangle rectangle = Rectangle.newBuilder()
                .setLo(Point.newBuilder().setLatitude(lowLat).setLongitude(lowLon).build())
                .setHi(Point.newBuilder().setLatitude(hiLat).setLongitude(hiLon).build())
                .build();
        //该调用与简单调用模式基本相似，只不过返回的是迭代对象。客户端可以用这个读取所有的返回对象。
        final Iterator<Feature> featureIterator = blockingStub.listFeatures(rectangle);
        for (int i = 1; featureIterator.hasNext(); i++) {
            final Feature feature = featureIterator.next();
            log.info("Feature {},value = {}", i, feature);
        }
    }

    /**
     * 异步、客户端流式调用
     * 将需要遍历的feature列表以流的方式发给服务端，服务端以流的方式接受请求消息，并处理消息。最后，处理完成所有消息后，将单个统计结果响应给客户端。
     * 这里我们需要用到异步stub
     */
    public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
        log.info("*** RecordRoute ***");
        final CountDownLatch finishLatch = new CountDownLatch(1);

        //新建一个StreamObserver（用来接受响应信息，此处基本上与服务端保持一致，只不过服务端写，这里读）
        StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {
            @Override
            public void onNext(RouteSummary summary) {
                log.info("Finished trip with {} points. Passed {} features. "
                                + "Travelled {} meters. It took {} seconds.",
                        summary.getPointCount(),
                        summary.getFeatureCount(),
                        summary.getDistance(),
                        summary.getElapsedTime());
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("RouteRecord Error", throwable);
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("Finished RecordRoute");
                //服务端响应完成后，放开门闩
                finishLatch.countDown();
            }
        };

        //新建一个StreamObserver作为客户端请求流，使用到异步stub，并指定该返回流（只不过在服务端，此流处理为只返回一条消息就completed）
        StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
        try {
            // 随机选择已存在的特征点位，作为列表发送给服务端
            for (int i = 0; i < numPoints; ++i) {
                int index = random.nextInt(features.size());
                Point point = features.get(index).getLocation();
                log.info("Visiting point {}, {}",
                        RouteGuideUtil.getLatitude(point),
                        RouteGuideUtil.getLongitude(point));
                // 将获取的点位放到请求流中
                requestObserver.onNext(point);
                // 让程序随机睡眠一小会，发送下一条记录。
                Thread.sleep(random.nextInt(1000) + 500);
                if (finishLatch.getCount() == 0) {
                    // 在发送请求的过程中，如果服务端抛异常，我看看到我们responseObserver中会接受到这个异常，并 finishLatch.countDown()
                    // 这个时候我们就要停止发送
                    return;
                }
            }
        } catch (RuntimeException e) {
            // Cancel RPC
            requestObserver.onError(e);
            throw e;
        }

        // 发送完所有点位后，结束请求流
        requestObserver.onCompleted();

        // Receiving happens asynchronously
        // 等待一分钟
        // 等待的是requestObserver发送完成后，服务端的处理时间（并不是等待request发送完）
        if (!finishLatch.await(1, TimeUnit.MINUTES)) {
            log.error("recordRoute can not finish within 1 minutes");
        }
    }

    /**
     * 双向流式调用
     * 每一次指定一个clientName，以作为记号，以便演示
     * 基本上跟客户端流式调用一样
     * 但是这里可以演示，服务端并不一定要等待客户端消息完成再响应
     * 而是来一条响应一条，可以看到客户端以边在发送消息，一边在接受服务端的响应
     * 两条流是独立的
     */
    public CountDownLatch routeChat(String clientName) {
        log.info("*** RouteChat ***");
        final CountDownLatch finishLatch = new CountDownLatch(1);
        StreamObserver<RouteNote> requestObserver =
                asyncStub.routeChat(new StreamObserver<RouteNote>() {
                    @Override
                    public void onNext(RouteNote note) {
                        log.info("{} Got message \"{}\" at {}, {}",
                                clientName,
                                note.getMessage(),
                                note.getLocation().getLatitude(),
                                note.getLocation().getLongitude());
                    }

                    @Override
                    public void onError(Throwable t) {
                        log.error("RouteChat Failed: {}", Status.fromThrowable(t));
                        finishLatch.countDown();
                    }

                    @Override
                    public void onCompleted() {
                        log.info("Finished RouteChat");
                        finishLatch.countDown();
                    }
                });

        try {
            RouteNote[] requests = {
                    newNote(clientName + " First message", 0, 0),
                    newNote(clientName + " Second message", 0, 10_000_000),
                    newNote(clientName + " Third message", 10_000_000, 0),
                    newNote(clientName + " Fourth message", 10_000_000, 10_000_000)};

            for (RouteNote request : requests) {
                log.info("Sending message \"{}\" at {}, {}",
                        request.getMessage(),
                        request.getLocation().getLatitude(),
                        request.getLocation().getLongitude());
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                requestObserver.onNext(request);
            }
        } catch (RuntimeException e) {
            // Cancel RPC
            requestObserver.onError(e);
            throw e;
        }
        // Mark the end of requests
        requestObserver.onCompleted();

        // return the latch while receiving happens asynchronously
        return finishLatch;
    }

    private RouteNote newNote(String message, int lat, int lon) {
        return RouteNote.newBuilder().setMessage(message)
                .setLocation(Point.newBuilder().setLatitude(lat).setLongitude(lon).build()).build();
    }

    public static void main(String[] args) throws InterruptedException {
        String target = "localhost:18080";
        List<Feature> features;
        try {
            features = RouteGuideUtil.parseFeatures(RouteGuideUtil.getDefaultFeaturesFile());
        } catch (IOException ex) {
            ex.printStackTrace();
            return;
        }
        ManagedChannel channel = ManagedChannelBuilder.forTarget(target).usePlaintext().build();
        try {
            RouteGuideClient client = new RouteGuideClient(channel);
            //  Looking for a valid feature
            // client.getFeature(409146138, -746188906);
            // Feature missing.
            // client.getFeature(0, 0);
            // Looking for features between 40, -75 and 42, -73.
            // client.listFeatures(400000000, -750000000, 420000000, -730000000);
            // Record a few randomly selected points from the features file.
            // client.recordRoute(features, 10);
            // Send and receive some notes.

            CountDownLatch finishLatchA = client.routeChat("ClientA");
            if (!finishLatchA.await(1, TimeUnit.MINUTES)) {
                log.error("routeChat can not finish within 1 minutes");
            }
            final CountDownLatch finishLatchB = client.routeChat("ClientB");
            if (!finishLatchB.await(1, TimeUnit.MINUTES)) {
                log.error("routeChat can not finish within 1 minutes");
            }
            final CountDownLatch finishLatchC = client.routeChat("ClientB");
            if (!finishLatchC.await(1, TimeUnit.MINUTES)) {
                log.error("routeChat can not finish within 1 minutes");
            }
        } catch (Exception e) {
            log.error("gRPC ERROR", e);
        } finally {
            channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
        }
    }
}
